package com.atguigu.watermark.generate;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

public class MyPuntuatedWatermarkGenerator<T> implements WatermarkGenerator<T> {
    private long maxTs;
    private final long delayTime;

    public MyPuntuatedWatermarkGenerator(long delayTime) {
        this.delayTime = delayTime;
        this.maxTs = Long.MIN_VALUE + delayTime - 1;
    }

    @Override
    public void onEvent(T t, long l, WatermarkOutput output) {
        long max = Math.max(maxTs, l);
        this.maxTs = max;
        output.emitWatermark(new Watermark(max - delayTime - 1));
        System.out.println("调用onEvent方法，获取目前为止的最大时间戳=" + maxTs+",watermark="+(maxTs - delayTime - 1));
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {

    }
}
